package com.heima.article.stream;

import com.alibaba.fastjson.JSON;
import com.heima.common.constants.HotArticleConstants;
import com.heima.model.mess.ArticleVisitStreamMess;
import com.heima.model.mess.UpdateArticleMess;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.kstream.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.time.Duration;

/**
 * kafka stream中间节点
 * 作用：
 * 接受上游发送的消息  {"articleId":"1606212835359199234", "type":LIKES,  "add":1}
 * 每10秒  按文章id聚合    {"articleId":1606212835359199234, "like":1, "view":0, "comment":0, "collect":0}
 * 发送到下游
 */
@Configuration
@Slf4j
public class HotArticleStreamProcessor {

    @Bean
    public KStream<String, String> kStream(StreamsBuilder streamsBuilder) {
        //1.监听上游topic
        KStream<String, String> stream = streamsBuilder.stream(HotArticleConstants.HOT_ARTICLE_SCORE_TOPIC);

        //2.聚合逻辑
        stream.map((key, value) -> {
            //key:null   value:{"articleId":"1606212835359199234", "type":LIKES,  "add":1}
            log.info("key: {}   value:{}", key, value);
            UpdateArticleMess mess = JSON.parseObject(value, UpdateArticleMess.class);
            //key:1606212835359199234       value:  LIKES:1
            return new KeyValue<>(mess.getArticleId().toString(), mess.getType() + ":" + mess.getAdd());
        })
                .groupByKey()//按文章id 聚合
                .windowedBy(TimeWindows.of(Duration.ofSeconds(10)))//时间窗口10s
                //.count()//统计10秒之内 每篇文章 有多少个行为操作
                .aggregate(new Initializer<String>() {//聚合方法
                               @Override
                               public String apply() {
                                   //聚合前数据初始化  在一个时间窗口之内第一次收到消息会初始化
                                   log.info("聚合之前的初始化方法");
                                   //{"articleId":null, "like":0, "view":0, "comment":0, "collect":0}
                                   return JSON.toJSONString(new ArticleVisitStreamMess());
                               }
                           },
                        new Aggregator<String, String, String>() {
                            /**
                             *  每次收到消息都会触发 apply方法做聚合
                             * @param key 文章的id 1606212835359199234
                             * @param value    LIKES:1
                             * @param aggregate  10秒之内上一次聚合的结果
                             *                   第一次调用格式类似于 {"articleId":null, "like":0, "view":0, "comment":0, "collect":0}
                             *                   非第一次调用格式类似于：{"articleId":1606212835359199234, "like":1, "view":0, "comment":0, "collect":0}
                             * @return
                             */
                            @Override
                            public String apply(String key, String value, String aggregate) {
                                log.info("聚合逻辑 key:{}  value:{}  aggregate:{}", key, value, aggregate);
                                ArticleVisitStreamMess mess = JSON.parseObject(aggregate, ArticleVisitStreamMess.class);
                                if(mess.getArticleId()==null){
                                    mess.setArticleId(Long.valueOf(key));
                                }
                                //根据value的内容   LIKES:1   将 mess对象相应的属性的值  累加
                                String[] split = value.split(":");// ["LIKES", "1"]
                                /*if(split[0].equals("LIKES")){
                                    mess.setLike(mess.getLike()+Integer.valueOf(split[1]));
                                }
                                if(split[0].equals("VIEWS")){
                                    mess.setView(mess.getView()+Integer.valueOf(split[1]));
                                }*/
                                switch (UpdateArticleMess.UpdateArticleType.valueOf(split[0])){
                                    case LIKES:
                                        mess.setLike(mess.getLike() + Integer.valueOf(split[1]));
                                        break;
                                    case VIEWS:
                                        mess.setView(mess.getView() + Integer.valueOf(split[1]));
                                        break;
                                    case COMMENT:
                                        mess.setComment(mess.getComment() + Integer.valueOf(split[1]));
                                        break;
                                    case COLLECTION:
                                        mess.setCollect(mess.getCollect() + Integer.valueOf(split[1]));
                                        break;
                                }
                                log.info("本次消息处理完成  累加结果:{}", JSON.toJSONString(mess));
                                return JSON.toJSONString(mess);
                            }
                        },
                        Materialized.as("abc"))//聚合
                .toStream()//转换KStream流
                .map((key, value) -> new KeyValue<>(key.key().toString(), value))//将聚合之后的结果封装到KeyValue对象
                //3.聚合完发送下游
                .to(HotArticleConstants.HOT_ARTICLE_INCR_HANDLE_TOPIC);//发送聚合后的消息到下游  格式{"articleId":1606212835359199234, "like":1, "view":0, "comment":0, "collect":0}


        return stream;
    }

}
